package com.mate.cloud.rocketmq.consumer;

import com.mate.cloud.constants.OrderConstant;
import com.mate.cloud.domian.OrderMessage;
import com.mate.cloud.domian.OrderMessageDTO;
import com.mate.cloud.enums.OrderActionEnum;
import com.mate.cloud.mapper.OrderMessageMapper;
import com.mate.cloud.rocketmq.CompensationService;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.concurrent.*;

/**
 * 消费者配置（多线程顺序消费）
 *
 * @author: MI
 * @email: 448341911@qq.com
 * @createTime: 2025/8/12 16:06
 * @updateUser: MI
 * @updateTime: 2025/8/12 16:06
 * @updateRemark: 修改内容
 * @version: 1.0
 */
@Slf4j
@Service
public class OrderConsumer {


    @Autowired
    private OrderMessageMapper orderMessageMapper;
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    @Autowired
    private CompensationService compensationService;

    // 线程池配置（按orderId哈希分片）
    private final ExecutorService threadPool = Executors.newFixedThreadPool(8);

    private final Map<String, LinkedBlockingQueue<OrderMessageDTO>> threadQueues = new ConcurrentHashMap<>();

    // 注册顺序消费监听器
    @RocketMQMessageListener(topic = "order-topic", consumerGroup = "${rocketmq.consumer.group}", consumeMode = ConsumeMode.ORDERLY  // 顺序消费模式
    )
    public class OrderMessageListener implements RocketMQListener<OrderMessageDTO> {
        @Override
        public void onMessage(OrderMessageDTO message) {
            // 按orderId哈希分配到线程
            String orderId = message.getOrderId();
            int threadId = Math.abs(orderId.hashCode()) % 8;
            threadQueues.computeIfAbsent(String.valueOf(threadId), k -> new LinkedBlockingQueue<>()).offer(message);
        }
    }

    // 初始化线程池消费任务
    @PostConstruct
    public void startConsumers() {
        for (int i = 0; i < 8; i++) {
            String threadId = String.valueOf(i);
            threadPool.submit(() -> {
                while (true) {
                    try {
                        OrderMessageDTO message = threadQueues.get(threadId).take();
                        processMessage(message); // 同一orderId的消息串行处理
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
            });
        }
    }

    // 消息处理逻辑（含异常重试和补偿）
    private void processMessage(OrderMessageDTO message) {
        String orderId = message.getOrderId();
        String orderKey = String.format(OrderConstant.order_id, orderId);
        try {
            // 1. 检查幂等性（Redis去重）
            if (Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(orderKey, "PROCESSING", 1, TimeUnit.HOURS))) {
                // 2. 业务处理（使用枚举消除硬编码）
                OrderMessage order = orderMessageMapper.selectById(orderId);
                if (order == null) {
                    log.error("订单不存在，处理失败: {}", orderId);
                    return;
                }

                // 通过枚举解析动作并设置状态（替代switch）
                OrderActionEnum actionEnum = OrderActionEnum.getByAction(message.getAction());
                order.setStatus(actionEnum.getTargetStatus());

                orderMessageMapper.insert(order);
                log.info("处理订单消息成功: {}", orderId);
            } else {
                log.warn("订单已处理，忽略重复消息: {}", orderId);
            }
        } catch (IllegalArgumentException e) {
            // 处理不支持的订单动作
            log.error("非法订单动作: {}", message.getAction(), e);
        } catch (Exception e) {
            log.error("处理订单消息失败: {}", orderId, e);
            // 可在此处添加重试逻辑（如发送到死信队列）
        } finally {
            // 注意：此处删除锁可能导致并发问题，建议使用Redis分布式锁的正确释放方式
            redisTemplate.delete("order:" + orderId);
        }
    }


}
